package highlevel;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import org.apache.kafka.streams.state.KeyValueStore;

import java.util.Arrays;
import java.util.Properties;

/**
 * @Author:RenPu
 * @Date: 2020/4/17 10:41
 * @Version: 1.0
 * @description:kafka streaming highlevel 有状态转换因子（数据恢复等）的使用
 */
public class HighLevelTransationStateful {
    

        public static void main(String[] args) {
          //创建properties对象配置kafka集群以及相关
            Properties properties = new Properties();
            properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092");
            properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());
            properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"word-dsl");
            properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG,3);
            

            //创建StreamBuild对象，可以获取kafkastreaming的topology任务
            StreamsBuilder streamsBuilder = new StreamsBuilder();

            
            //从消费主题进行获取record中的数据
            KStream<String, String> kStream= streamsBuilder.stream("t9");


            //有状态转换因子count,作用指定存储结构，来存储k,v数据结构，统计相同的key，出现的record的次数
//            KTable<String, Long> kTable = kStream.flatMapValues(value ->
//                    Arrays.asList(value.split(" ")))
//
//                    //根据value的值进行分组
//                    .groupBy((k, v) -> v)
//                    .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("c158").withKeySerde(Serdes.String()).withValueSerde(Serdes.Long()));
//            kTable.toStream().print(Printed.toSysOut());



            //有状态聚合（aggregate）转换因子
//            KTable<String, Long> kTable = kStream.flatMapValues(value ->
//                    Arrays.asList(value.split(" ")))
//
//                    //根据value的值进行分组
//                    .groupBy((k, v) -> v)
//
//                    //聚合：第一个参数聚合的初始值从0开始，第二个参数聚合逻辑，第三个参数【必须】指定状态存储k,v的数据类型
//                    .aggregate
//                            (
//                            ()->0L,
//                            (k,v,agg)->agg+1L,Materialized.<String,Long,KeyValueStore<Bytes, byte[]>>as("160")
//                                            .withKeySerde(Serdes.String())
//                                            .withValueSerde(Serdes.Long())
//                            );
//
//            kTable.toStream().print(Printed.toSysOut());


            //reduce 规约计算 有状态转换因子

            KTable<String, Long> kTable = kStream.flatMapValues(value ->
                    Arrays.asList(value.split(" ")))

                    /*
                    更改前： k:null v:hello
                            k:null v:word
                    更改后：k:hello v:1L
                            k:word v:1L
                     手动指定k,v的数据类型
                     原因：便于reduce进行对v进行计算
                    */
                    .map((k,v)->new KeyValue<String,Long>(v,1L))

                     //替换默认的String,k,v类型
                    .groupByKey(Grouped.with(Serdes.String(),Serdes.Long()))

                    //.groupByKey()//这个会出现error，一个是属于分组存储groupByKey()，一个是转存存储reduce((v1,v2)，二个阶段的存储都需要指定k,v的数据类型

                    //聚合：第一个参数聚合的初始值从0开始，第二个参数聚合逻辑，第三个参数【必须】指定状态存储k,v的数据类型
                    .reduce((v1,v2)->v1+v2,
                            Materialized.<String,Long,KeyValueStore<Bytes, byte[]>>as("c160").withKeySerde(Serdes.String()).withValueSerde(Serdes.Long()));

            kTable.toStream().print(Printed.toSysOut());



















            //获取topology任务对象
            Topology topology = streamsBuilder.build();

            //创建kafkaStream对象启动任务
            KafkaStreams kafkaStreams = new KafkaStreams(topology,properties);

            //开启任务
            kafkaStreams.start();


        }





}
